package cn.yj.frame.proxy;

import cn.yj.frame.netty.ResponseHandler;
import cn.yj.frame.protrol.RpcProtocol;
import cn.yj.register.RegisterConfig;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.serialization.ClassResolvers;
import io.netty.handler.codec.serialization.ObjectDecoder;
import io.netty.handler.codec.serialization.ObjectEncoder;
import io.netty.handler.timeout.IdleStateHandler;
import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.util.concurrent.TimeUnit;

/**
 * Rpc调用方代理对象
 *
 * @author 永健
 * @since 2022-01-18 17:11
 */
public class RpcReferenceProxy implements InvocationHandler {

	private Class<?> aClass;

	private String beanName;

	public RpcReferenceProxy(Class<?> aClass, String beanName) {
		this.beanName = beanName;
		this.aClass = aClass;
	}

	@Override
	public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
		ResponseHandler responseHandler = new ResponseHandler();
		EventLoopGroup group = new NioEventLoopGroup(1);
		Bootstrap bootstrap =
			new Bootstrap()
				.group(group)
				.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 5000)
				.channel(NioSocketChannel.class)
				.handler(new ChannelInitializer<SocketChannel>() {
					@Override
					protected void initChannel(SocketChannel socketChannel) throws Exception {
						// 心跳监测机制
						socketChannel.pipeline().addLast(new IdleStateHandler(3, 3, 3, TimeUnit.SECONDS));
						// 编码处理器
						socketChannel.pipeline().addLast("encoder", new ObjectEncoder());
						// 解码处理器
						socketChannel.pipeline().addLast("decoder", new ObjectDecoder(1024,
							ClassResolvers.cacheDisabled(this.getClass().getClassLoader())));
						// 消息处理器
						socketChannel.pipeline().addLast(responseHandler);
					}
				});

		// 向注册中心拉取服务列表
		String ip = RegisterConfig.getInstance().loadBalanceIp(this.beanName);
		checkIsNoExits(ip);

		// 连接提供者
		ChannelFuture connect = bootstrap.connect(ip.split(":")[0], Integer.parseInt(ip.split(":")[1])).sync();
		Channel channel = connect.channel();

		// 包装协议对象
		RpcProtocol rpcProtocol = new RpcProtocol();
		rpcProtocol.setMethodName(method.getName());
		rpcProtocol.setMethodParams(args);
		rpcProtocol.setMethodParamsType(method.getParameterTypes());
		rpcProtocol.setBeanName(this.beanName);
		// 传输
		channel.writeAndFlush(rpcProtocol);
		channel.closeFuture().sync();
		return responseHandler.getResult();
	}

	private void checkIsNoExits(String ip) {
		if (ip == null || "".equals(ip)) {
			throw new RuntimeException("服务不存在:" + this.beanName);
		}
	}
}
